理解 Node.js 中的 Stream

Node.js 中的流以难以使用而著称，甚至更难理解。

用 Dominic Tarr 的话来说：“stream 是 node 中最好用的也是最不易理解的存在。即使是 Redux 的创建者和 React.js 的核心团队成员 Dan Abramov 也害怕 node stream。

本文将帮助您理解 stream 以及如何使用它们。所以，不要害怕。follow me ! 🏖

Stream 是支持 Node.js 应用程序的基本概念之一。它们是一种数据处理方法，用于依次读写数据的输入并输出。

stream 是一种有效地处理读/写文件、网络通信或任何类型的端到端信息交换的方法。

stream 的独特之处在于，它不像传统程序那样 一次性 将文件读入内存，而是逐块读取数据块，处理其内容而不会将其全部保存在内存中。

这使得 stream 在处理大量数据时非常强大。例如，文件大小可能大于空闲内存空间，因此不可能将整个文件读取到内存中以进行处理。这就是 stream 来拯救我们的地方！

使用 stream 来处理较小的数据块，使它可以读取整体较大的文件。

让我们以 Youtube 或 Netflix 等 ”流媒体“ 服务为例：这些服务不会让你一次性下载整个视频和音频。相反，你的浏览器会接收到连续不断的视频片段，让接收人立即开始观看。

然而，stream 不仅仅与媒体或大数据打交道。它们还为我们的代码提供了 ”可组合性“ 的能力。考虑到可组合性的设计意味着几个组件可以以某些方式组合，从而产生相同类型的结果。在 node.js 中，可以使用 stream 将数据传输到各个较小的代码片段中，它们可以自由携带与组合，从而编写出功能强大的代码片段。

与其他处理数据方法相比，stream 提供了两个主要优势：

如果你使用过 nodejs，那么你可能遇到过 stream。例如，在基于 nodejs 的 http 服务器中，request 是可读流，response 是可写流。你可能已经使用了 fs 模块，它允许你处理可读和可写的文件流。在你使用 express，你都在使用流与客户端交互，同时，你使用的每个数据库连接驱动程序中都有应用流，因为 TCP 套接字，TLS 堆栈和其他连接都是基于 nodejs 的。

我们首先引入 Readable，然后对它进行初始化。

现在 stream 已经初始化，我们可以发送数据给它：

强烈建议在处理流时使用 async iterator（异步迭代器）。根据 Dr. Axel Rauschmayer 的说法，异步迭代是一种异步检索数据容器内容的协议（意味着当前的 ”任务“ 可能会在检索项之前暂停）。同样，值得一提的是，stream async iterator 的实现使用的内部的 ”可读“ 事件。

你可以使用 async iterator 读取可读流：

它也可以收集可读流的内容到一个字符串：

一定要记住，不要将 async 函数与 EventEmitter 混合使用，因为当前，当 reject 在事件处理程序中 emit 时，没有办法捕获它，这导致很难跟踪错误和内存泄漏。当前的最佳实践是始终将异步函数的内容包装在 try/catch 块中并处理错误，但这很容易出错。这个 pull request 旨在解决这个问题，可能已经合入到 node 的核心代码中了。

关于 node stream 的更多异步遍历解析，请阅读[这篇文章])(2ality.com/2019/11/nod…)

stream.Readable.from(iterable, [options]) 它是一种常用方法，用于从迭代器中创建可读流，它保存 iterable 中包含的数据。Iterable 可以是同步迭代，也可以是异步迭代。参数选项是可选的，可以用来指定文本编码。

根据 Streams API，可读流有效地在两种模式中操作： flowing 和 paused。可读流可以是对象模式或非对象模式，而不管它是流模式还是暂停模式。

flowing mode，数据自动从系统底层读取，并通过 EventEmitter 接口事件尽可能快地提供给应用程序。

paused mode，stream.read() 方法必须显示调用，以从流中读取数据块。

在 flowing 模式中，要从流中读取数据，可以监听 data 事件并附加回调。当数据块可用时，可读流再发送一个 end 事件并执行回调。如下示例：

fs.createReadStream() 创建了一个可读流。起初，stream 处于静止状态。**一旦监听 data 事件并附加回调，它就开始流动。**在那之后，数据块被读取并传递给你的回调。流实现者决定 data 事件触发的频率。例如， http 请求可能会在每读取几 kb 数据时就触发一个 data 事件。当你从文件读取数据时，你可以决定在读取一行后触发一个 data 事件。

当没有更多的数据要读取（到达 end ）时，流触发一个 end 事件。在上面的代码片段中，我们监听这个事件，以便在到达终点时得到通知。

此外，如果有错误，流将触发 error 事件。

在 paused 模式中，你只需要在流实例上反复调用 read()，直到每个数据块都被读取，如下示例:

read 函数的作用是：从内部缓冲区中读取一些数据并返回这些数据。当没有要读取的内容时，它返回 null。在 while 循环中，我们检查 null 值并终止循环。请注意，当可以从流中读取数据块时，将触发 readable 事件。

所有可读流可以从 paused 模式开始，通过以下任意一种方式切换为 flowing 模式：

可读流同样可以通过以下任一种方式切换回 paused 模式：

需要记住的重要概念是，只有提供了使用或忽略数据的机制（事件），可读流才会生成数据。如果消费机制被禁用或取消，可读流将尝试停止生成数据。可读流可暂停，可继续，可通过 readable.read() 逐步读取数据。如果 “readable” 事件处理程序被删除，那么如果存在 “data” 事件处理程序，流将再次开始流动。

要将数据写入可写流，需要在流实例上调用 write()。如下示例：

上面的代码很简单。它只是从输入流中读取数据块，然后使用 write() 将数据写入目标。此函数返回一个布尔值，指示操作是否成功。如果为 true，那么写入操作成功，您可以继续写入更多数据。如果返回 false，则意味着出现了错误，此时您无法编写任何内容。可写流将通过触发 drain 事件让你知道何时可以继续开始写入数据。

调用 writable.end() 方法表明不会再向可写流写入数据。调用后，其可选的回调函数将作为 finish 事件触发。

使用可写流，你可以从可读流读取数据:

（推荐）你还可以使用异步迭代器来写入可写入流：

finish() 的默认是基于回调的，但是可以通过 util.promisify() (第A行) 将其转换为基于 Promise 的版本。

在本例中，它使用了以下两种模式：

在处理写入错误时，等待 drain 事件（可写入）继续写入可写流 （行 B）：

写入完成后，关闭可写流（行 C）：

pipeline 是一种机制，我们将一个流的输出作为另一个流的输入。它通常用于从一个流获取数据，并将该流的输出传递给另一个流。管道操作没有限制。换句话说，管道用于在多个步骤中处理流数据。

⚠️注意！pipe 或 pipeline只有目标流为 Duplex 流或 Transform 流才可以形成管道链。文档

在 Node 10.x 引入了 stream.pipeline()。这是一个模块方法，用于在流之间进行管道传输，转发错误，适当清理，并在管道完成时提供回调。

如下就是使用 pipeline 的示例：

pipeline 应该替代 pipe ，因为 pipe 不安全。

nodejs stream 模块提供了构建所有流 api 的基础。

stream 模块是 Node.js 中默认提供的本地模块。stream 是 EventEmitter 类的一个实例，它在 Node 中异步地处理事件。因此，stream 本质上是基于事件的。

获取 stream 模块：

stream 模块对于创建新类型的流实例非常有用。通常没有必要使用 stream 模块来创建流，还有更常用的其他方式。

由于其优势，许多 Node.js 核心模块提供了本地流处理能力，最明显的是：

下面是与可写流相关的重要事件：

这都是关于 stream 的基础知识。streams、pieps和 chaining 是 Node.js 的核心和最强大的特性。 streams 确实可以帮助你编写整洁和高性能的代码来执行 I/O。

此外，还有一个值得关注的 Node.js 战略计划，名为 BOB，旨在改进 Node.js stream 数据接口，既包括在 Node.js 核心内部，也有希望成为未来的公共 api。